package cn.jly.flink.window;

import cn.jly.flink.entity.SensorReading;
import cn.jly.flink.source2sink.datastream.source.CustomSource;
import cn.jly.flink.utils.FlinkUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Comparator;

/**
 * https://gitbook.cn/gitchat/column/5dad4a20669f843a1a37cb4f/topic/5db6a47df6a6211cb9616512
 *
 * 窗口的开始时间是按照 TimeWindow 类的getWindowStartWithOffset方法计算，参数单位都是ms，windowSize是窗口长度
 * public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) {
 *         return timestamp - (timestamp - offset + windowSize) % windowSize;
 * }
 *
 * @PackageName cn.jly.flink.window
 * @ClassName SlidingWindowDemo
 * @Description 滑动窗口测试
 * @Author 姬岚洋
 * @Date 2021/1/29 下午 2:53
 */
public class SlidingWindowDemo {
    /**
     * 最高温度
     */
    private static final String HIGHEST_TEMPERATURE_TAG_NAME = "highest_temperature_tag";
    private static final String LOWEST_TEMPERATURE_TAG_NAME = "lowest_temperature_tag";

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = FlinkUtils.getStreamExecutionEnv();

        // 高温tag
        final OutputTag<Tuple2<String, Double>> highestTempTag = new OutputTag<Tuple2<String, Double>>(HIGHEST_TEMPERATURE_TAG_NAME) {
        };
        // 低温tag
        final OutputTag<Tuple2<String, Double>> lowestTempTag = new OutputTag<Tuple2<String, Double>>(LOWEST_TEMPERATURE_TAG_NAME) {
        };

        final SingleOutputStreamOperator<Tuple2<String, Double>> process = env.addSource(new CustomSource.CustomSensorReadingSourceFunction("温度生成"))
                // 允许3秒的延迟数据
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<SensorReading>forBoundedOutOfOrderness(Duration.of(3, ChronoUnit.SECONDS))
                                .withTimestampAssigner(new SerializableTimestampAssigner<SensorReading>() {
                                    @Override
                                    public long extractTimestamp(SensorReading element, long recordTimestamp) {
                                        return element.getTimestamp();
                                    }
                                })
                )
                .keyBy(new KeySelector<SensorReading, String>() {
                    @Override
                    public String getKey(SensorReading value) throws Exception {
                        return value.getId();
                    }
                })
                // 每5秒统计过去10秒数据
                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                // 允许迟到数据，在assignTimestampsAndWatermarks计算窗口触发计算之后不关闭窗口，
                // 继续等5s钟，且来一条数据就和之前的窗口进行process操作
                // .allowedLateness(Time.seconds(5))
                // 将最终迟到的数据输出到一个侧输出流，要之后获取的话，需要把OutputTag提取出来，之后将迟到的数据与先前的结果聚合处理，以得到最终正确的结果
                // .sideOutputLateData(new OutputTag<SensorReading>("late_data"){})
                .process(new ProcessWindowFunction<SensorReading, Tuple2<String, Double>, String, TimeWindow>() {
                    @Override
                    public void process(String key, Context context, Iterable<SensorReading> elements, Collector<Tuple2<String, Double>> out) throws Exception {
                        // 获取窗口信息
                        // context.window();

                        if (CollectionUtils.size(elements) > 0) {
                            double sumTemp = 0d;
                            final ArrayList<SensorReading> list = new ArrayList<>();

                            // 统计温度和
                            for (SensorReading next : elements) {
                                sumTemp += next.getTemperature();
                                list.add(next);
                            }

                            // 按照温度排序
                            list.sort(Comparator.comparingDouble(SensorReading::getTemperature));

                            // 正常输出窗口的平均温度
                            out.collect(Tuple2.of(key, sumTemp / CollectionUtils.size(elements)));
                            // 侧输出流输出窗口的最高和最低温度
                            context.output(highestTempTag, Tuple2.of(key, list.get(0).getTemperature()));
                            context.output(lowestTempTag, Tuple2.of(key, list.get(list.size() - 1).getTemperature()));
                        }
                    }
                });

        // 平均温度
        process.filter(new FilterFunction<Tuple2<String, Double>>() {
            @Override
            public boolean filter(Tuple2<String, Double> value) throws Exception {
                return "sensor_4".equals(value.f0);
            }
        }).print("每5秒统计过去10秒平均温度为");
        // 最高温度
        process.getSideOutput(highestTempTag).filter(
                new FilterFunction<Tuple2<String, Double>>() {
                    @Override
                    public boolean filter(Tuple2<String, Double> value) throws Exception {
                        return "sensor_4".equals(value.f0);
                    }
                }
        ).print("过去10秒的最高温度为");
        // 最低温度
        process.getSideOutput(lowestTempTag).filter(new FilterFunction<Tuple2<String, Double>>() {
            @Override
            public boolean filter(Tuple2<String, Double> value) throws Exception {
                return "sensor_4".equals(value.f0);
            }
        }).print("过去10秒的最低温度为");

        FlinkUtils.executeStream(env, "SlidingWindowDemo");
    }
}
